#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <stdlib.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <errno.h>

#include <cm/net/net_manager.h>
#include <cm/net/sock_read_write.h>
#include <cm/thread/thread_pool.h>
#include <cm/struct/list.h>

#include "net_constant.h"
#include "net_session.h"
#include "net_server.h"

struct net_manager {
    pthread_t tid;

    struct list_head servers;
    struct list_head clients;

    pthread_rwlock_t serverslock;
    pthread_rwlock_t clientslock;

    struct thread_pool_t *pool;

    int flag;
};

static void *net_manager_proc(void *arg);

/*
 * @brief	创建一个net manager, 用来添加网络端（服务端或客户端）
 * 		1、负责接收网络数据，包括接收新的连接(accept)
 * 		2、调用会话和服务器中关联的回调函数
 * @param	flag：选项，详见各个选项的定义
 * @return	成功则返回指向管理者的指针,失败则返回NULL
 */
struct net_manager* netm_create(int flag)
{
    struct net_manager *m = (struct net_manager*) calloc(1,
            sizeof(struct net_manager));

    if (m == NULL)
        return NULL;

    m->pool = thread_pool_create(4);
    if (m->pool == NULL) {
        free(m);
        return NULL;
    }

    init_list_head(&(m->servers));
    init_list_head(&(m->clients));

    pthread_rwlock_init(&(m->serverslock), NULL);
    pthread_rwlock_init(&(m->clientslock), NULL);

    m->flag = flag;

    if (0 != pthread_create(&(m->tid), NULL, net_manager_proc, m)) {
        pthread_rwlock_destroy(&(m->serverslock));
        pthread_rwlock_destroy(&(m->clientslock));
        thread_pool_destroy(m->pool);
        free(m);
        m = NULL;
    }

    return m;
}

/*
 * @brief	销毁一个net manager, 关闭其中所有的连接及服务器
 * @return	*manager将被设置为NULL，因为它不能再使用
 */
void netm_destroy(struct net_manager **m)
{
    if (m == NULL || *m == NULL)
        return;

    if ((*m)->tid != 0) {
        pthread_cancel((*m)->tid);
        pthread_join((*m)->tid, NULL);
        (*m)->tid = 0;
    }

    /* 释放锁后，其他地方不能再访问这两个链表 */
    release_rwlock((*m)->clientslock);
    close_sessions(&((*m)->clients));

    release_rwlock((*m)->serverslock);
    close_servers(&((*m)->servers));

    thread_pool_destroy((*m)->pool);

    free(*m);
    *m = NULL;
}

extern int inet4_tcp_server(const char *ip, unsigned short port);

/*
 * @brief	向网络管理者m中添加一个服务器
 * @param	ip, port: 指定服务器监听的地址和端口,ip设置NULL则监听所有地址
 * @param	rsize, wsize: 指定服务器将为每一个连接上来的会话创建多大的接收和发送缓存
 * @param	proc: 指定当服务器接收到数据时，使用哪个函数来处理
 * @return	服务器对象，失败则返回NULL
 */
struct net_server* netm_addserver(struct net_manager *m, const char *ip,
        unsigned short port, size_t rsize, size_t wsize, session_proc proc)
{
    if (m == NULL)
        return NULL;

    int fd = inet4_tcp_server(ip, port);
    if (fd == -1)
        return NULL;

    struct net_server *server = server_new(fd, rsize, wsize, proc);
    if (server == NULL) {
        close(fd);
        return NULL;
    }

    if (0 == pthread_rwlock_wrlock(&(m->serverslock))) {
        list_add_tail(&(server->list), &(m->servers));
        pthread_rwlock_unlock(&(m->serverslock));
    } else {
        printf("Can't add server, get lock failed.\n");
        server_delete(&server);
    }

    return server;
}

extern int inet4_tcp_client(const char *ip, unsigned short port);

/*
 * @brief	向m中添加一个客户端连接
 * @param	ip,port: 指定对端服务的地址和端口
 * @param	rsize, wsize: 指定该会话需要创建多大的接收和发送缓存，发送缓存可以为0
 * @param	proc: 指定当接收到数据时，使用哪个函数来处理数据
 * @return	建立连接后的会话
 */
struct net_session* netm_addclient(struct net_manager *m, const char *ip,
        unsigned short port, size_t rsize, size_t wsize, session_proc proc)
{
    if (m == NULL || ip == NULL)
        return NULL;

    int fd = inet4_tcp_client(ip, port);
    if (fd == -1)
        return NULL;

    struct net_session *session = session_new(fd, SESSION_CLIENT, rsize, wsize,
            proc);
    if (session == NULL) {
        close(fd);
        return NULL;
    }

    if (0 == pthread_rwlock_wrlock(&(m->clientslock))) {
        list_add_tail(&(session->list), &(m->clients));
        pthread_rwlock_unlock(&(m->clientslock));
    } else {
        printf("Can't add session, get lock failed.\n");
        session_delete(&session);
    }

    return session;
}

/*
 * @brief	关闭一个服务端，将它从manager中移除，同时关闭其中所有会话
 * @return	关闭失败返回-1（可能服务器正在处理数据），成功返回0
 */
int netm_closeserver(struct net_manager *manager, struct net_server **server)
{
    if (manager == NULL || server == NULL || *server == NULL)
        return -1;

    if (0 != pthread_rwlock_wrlock(&(manager->serverslock)))
        return -1;

    server_delete(server);

    pthread_rwlock_unlock(&(manager->serverslock));

    return 0;
}

/*
 * @brief	关闭一个客户端，将它从manager中移除
 * @return	关闭失败返回-1（可能正在处理数据），成功返回0
 */
int netm_closesession(struct net_manager *manager, struct net_session **session)
{
    if (manager == NULL || session == NULL || *session == NULL) {
        printf(
                "netm_closesession: param is null, manager = %p, session = %p, * session = %p\n",
                manager, session, *session);
        return -1;
    }

    switch ((*session)->type) {
    case SESSION_CLIENT:
        if (0 != pthread_rwlock_wrlock(&(manager->clientslock))) {
            return -1;
        }
        session_delete(session);
        pthread_rwlock_unlock(&(manager->clientslock));
        break;
    case SESSION_SERVER:
        /* 这里准确的应该是加锁会话所在的server,
         * 不过操作server中会话链表时，会先加锁这个serverslock */
        if (0 != pthread_rwlock_wrlock(&(manager->serverslock))) {
            return -1;
        }
        session_delete(session);
        pthread_rwlock_unlock(&(manager->serverslock));
        break;
    default:
        printf("\tnetm_closesession, error session type.\n");
        break;
    }

    return 0;
}

/*
 * @brief	将数据通过session发送出去
 * @param	session：一个会话的指针，一般是通过netm_addclient或在回调函数session_proc中获得
 * 		data，size：要发送的数据位置和大小
 * 		timeout：超时设置，0表示不设置超时，可以永远阻塞。 >0的值设置超时时间，单位毫秒(千分之一秒)
 * @return	>0表示一共发送出去的数据大小, =0 表示超时  <0表示出错
 */
enum SESSION_SEND_RESULT session_send(struct net_session *session,
        const char *data, size_t size, int timeout)
{
    if (session == NULL || data == NULL || size <= 0)
        return SSR_SUCCESS;

    enum SESSION_SEND_RESULT ret = SSR_SUCCESS;

    if (0 != lock_session(session)) {
        perror("lock session failed: ");
        return SSR_ERROR;
    }

    /* TIMEOUT BROKEN ERROR状态的会话不进行发送 */
    if (session->state >= SESSION_TIMEOUT) {
        printf("Wrong session(state = %d).\n", session->state);
        unlock_session(session);
        return SSR_ERROR;
    }

    if (session->write_buffer != NULL) {
        if (0 != cmfifo_putdata(session->write_buffer, data, size)) {
            session->buf_full_count++;
            printf("\t=>>> put into write buffer failed. full count = %d\n",
                    session->buf_full_count);
            if (session->buf_full_count >= SESSION_MAX_FULL_TIME) {
                printf("\t ====> full time >= 100\n");
                ret = SSR_ERROR;
            } else {
                ret = SSR_FULL;
            }
        } else {
            session->buf_full_count = 0;
        }
    } else {
        int r = send_nbytes(session->fd, data, size, timeout);
        if (r == 0) {
            session->state = SESSION_TIMEOUT;
            ret = SSR_TIMEOUT;
            printf("\t=> session_send && no write buffer && timeout.\n");
        } else if (ret < 0) {
            session->state = SESSION_ERROR;
            ret = SSR_ERROR;
            printf("\t=> session_send && no write buffer && error.\n");
        }
    }

    unlock_session(session);
    return ret;
}

/*
 * @brief	将数据发送给服务端中的所有会话
 * @param	server：一个服务端的指针，一般是通过netm_addserver获得
 * 		data，size：要发送的数据位置和大小
 * 		timeout：超时设置，0表示不设置超时，可以永远阻塞。 >0的值设置超时时间，单位毫秒(千分之一秒)
 * @return	>= 0 表示一共发给了多少个客户端会话, < 0 表示出错
 */
int server_send(struct net_server *server, const char *data, size_t size,
        int timeout)
{
    if (server == NULL || data == NULL || size <= 0)
        return SSR_SUCCESS;

    int count = 0;
    struct net_session *s = NULL;
    enum SESSION_SEND_RESULT ret = SSR_SUCCESS;

    if (0 != lock_server(server)) {
        perror("lock server failed.\n");
        return -1;
    }

    struct list_head *pos = NULL;
    struct list_head *safe = NULL;
    list_for_each_safe(pos, safe, &(server->sessions))
    {
        s = list_entry(pos, struct net_session, list);
        ret = session_send(s, data, size, timeout);
        switch (ret) {
        case SSR_ERROR:
        case SSR_TIMEOUT:
            perror("server send failed, delete session.\n");
            session_delete(&s); /* 其所在的服务器已经上锁，因此可以删除 */
            break;
        case SSR_FULL:
//			if (s->timeout_count > 100) {
//				printf(
//						"server send failed, buffer full(100).\n");
//				session_delete(&s);
//			}
            break;
        default:
            count++;
            break;
        }
    }

    unlock_server(server);
    return count;
}

/*------------------ USER API END --------------------*/

static void _do_read_write(struct net_manager *m);

/*
 * @brief	net manager的主线程,负责读取网络数据
 * @param	arg: 传入的是该线程所对应的net_manager
 */
static void *net_manager_proc(void *arg)
{
    struct net_manager *manager = (struct net_manager*) arg;

    while (1) {
        _do_read_write(manager);

		usleep(1000*10);
	}

    return NULL;
}

/*--------------- SET FD_SET BEGIN----------------*/

static inline void _fd_set(fd_set *set, int *max_fd, int fd)
{
    if (fd < 0)
        return;

    FD_SET(fd, set);
    if (fd > *max_fd)
        *max_fd = fd;
}

/* 将[一串]会话的fd设置到set中 */
static inline void set_sessions_fd(fd_set *rfds, fd_set *wfds, int *mf,
        struct list_head *s)
{
    struct list_head *pos;
    list_for_each(pos, s)
    {
        struct net_session *ns = list_entry(pos, struct net_session, list);
        if (0 != trylock_session(ns)) {
            printf("Try lock session for read failed, continue.\n");
            continue;
        }

        _fd_set(rfds, mf, ns->fd);
        if (ns->write_buffer != NULL)
            _fd_set(wfds, mf, ns->fd); /* 有发送缓冲的才加入此fd */

        unlock_session(ns);
    }
}

/* 设置[一串]server的监听套接字以及server中所有会话的套接字 */
static inline void set_server_fd(fd_set *rfds, fd_set *wfds, int *max_fd,
        struct list_head *s)
{
    struct list_head *pos;
    list_for_each(pos, s)
    {
        struct net_server *ns = list_entry(pos, struct net_server, list);

        if (0 != lock_server(ns))
            continue;

        _fd_set(rfds, max_fd, ns->listen_fd); /* 监听套接字只需要读 */
        set_sessions_fd(rfds, wfds, max_fd, &(ns->sessions));

        unlock_server(ns);
    }
}

static void set_rw_fds(fd_set *rfds, fd_set *wfds, int *max_fd,
        struct net_manager *manager)
{
    if (manager == NULL || rfds == NULL || wfds == NULL || max_fd == NULL)
        return;

    FD_ZERO(rfds);
    FD_ZERO(wfds);
    *max_fd = -1;

    pthread_rwlock_rdlock(&(manager->serverslock));
    set_server_fd(rfds, wfds, max_fd, &(manager->servers));
    pthread_rwlock_unlock(&(manager->serverslock));

    pthread_rwlock_rdlock(&(manager->clientslock));
    set_sessions_fd(rfds, wfds, max_fd, &(manager->clients));
    pthread_rwlock_unlock(&(manager->clientslock));

    return;
}

/*--------------- SET FD_SET END----------------*/

/*
 * @brief	用于线程池中处理任务的函数
 * @param	参数都是一个session的指针
 * 		这个函数根据session的不同状态调用session中的回调函数
 */
static inline void * _users_work(void *arg)
{
    struct net_session *temp = (struct net_session *) arg;
    if (temp->state == SESSION_BROKEN || temp->state == SESSION_ERROR) {
        temp->proc(SESSION_E_BROKEN, temp, NULL);
    } else if (temp->state == SESSION_INIT) {
        temp->proc(SESSION_E_NEW, temp, NULL);
        temp->state = SESSION_NONE;
    } else if (temp->state == SESSION_DATA) {
        temp->proc(SESSION_E_DATA, temp, temp->read_buffer);
        temp->state = SESSION_NONE;
    } else {
        printf("Session state should not call user's proc.\n");
    }

    return NULL;
}

/*--------------- READ FOR SESSION OR SERVER BEGIN ---------------*/
static int _do_session_read(fd_set *rset, struct net_session *temp,
        struct net_manager *net)
{
    if (0 != trylock_session(temp))
        return 0;

    if (temp->fd < 0) {
        unlock_session(temp);
        return -1;
    }

    if (!FD_ISSET(temp->fd, rset)) {
        unlock_session(temp);
        return 0;
    }

    char buf[1024 * 1024];

    int flags = fcntl(temp->fd, F_GETFL, 0);
    fcntl(temp->fd, F_SETFL, flags | O_NONBLOCK);

    int ret = read(temp->fd, buf, 1024 * 1024); // TODO 一次只读了1024
    fcntl(temp->fd, F_SETFL, flags);

    int r = 0;
    if (ret > 0) {
        if (0 != cmfifo_putdata(temp->read_buffer, buf, ret)) {
            printf("buffer full.\n");
            //TODO 缓存满了
        }
        temp->state = SESSION_DATA;
    } else {
        if (ret == 0) {
            printf("connection broken.\n");
//			temp->state = SESSION_BROKEN;
        } else {
            perror("read: ");
//			temp->state = SESSION_ERROR;
        }
        /* 如果是服务端中的会话断开了，则删掉它，因为他是我们内部创建的
         TODO BUG 这里可能有遗留数据没处理 */
        if (temp->type == SESSION_SERVER) {
            unlock_session(temp);
            session_delete(&temp);
            return -1;
        } else {
            close(temp->fd);
            temp->fd = -1;
            temp->state = SESSION_BROKEN;
        }
        r = -1;
    }
    unlock_session(temp); /* 解锁以后再调用用户的回调，避免死锁 */

    // 此时可能持有的锁：manager的server锁或clients锁 某个server的锁
    // 因此 如果用户函数调用其它netm相关函数需要使用这些锁，则会死锁
    // 解决方案：
    // 0、回调函数中不要使用netm相关函数
    // 1、新建一个线程来执行用户回调，还可以避免回调中的阻塞
    // 2、更好的方案：线程池  [DONE]
    thread_pool_add_work(net->pool, _users_work, temp);
    return r;
}

static inline void _do_session_write(fd_set *wset, struct net_session *s,
        struct net_manager *net)
{
    if (0 != trylock_session(s))
        return;

    if (!FD_ISSET(s->fd, wset) || s->write_buffer == NULL
            || cmfifo_empty(s->write_buffer)) {
        unlock_session(s);
        return;
    }

    int flags = fcntl(s->fd, F_GETFL, 0);
    fcntl(s->fd, F_SETFL, flags | O_NONBLOCK);

    char* ptr[2] = { NULL, NULL };
    unsigned size[2] = { 0, 0 };
    cmfifo_dataptr(s->write_buffer, &ptr[0], &size[0], &ptr[1], &size[1]);
    int i;
    for (i = 0; i < 2 && size[i] > 0 && ptr[i] != NULL; i++) {
        int done = write(s->fd, ptr[i], size[i]);
//		printf("session write: total = %u, actually =  %d\n", size[i],
//				done);
        if (done > 0)
            cmfifo_setstatus(s->write_buffer, CMFIFO_GET, done);
        if (done < (int) size[i])
            break;
    }

    fcntl(s->fd, F_SETFL, flags);

    unlock_session(s);
}

/* 处理会话的可读，将数据拷贝到会话的缓存中 */
static inline void _do_session_rw(fd_set *rset, fd_set *wset,
        struct list_head *s, struct net_manager *net)
{
    struct list_head *pos = NULL;
    struct list_head *safe = NULL;
    list_for_each_safe(pos, safe, s)
    {
        struct net_session *temp = list_entry(pos, struct net_session, list);
//		printf("session == %p, net = %p\n", temp, net);
        if (0 == _do_session_read(rset, temp, net))
            _do_session_write(wset, temp, net);

    }
}

/* 当服务器的监听套接字可读时，处理新到来的连接 */
static inline void _do_server_accept(struct net_server *ns,
        struct net_manager* net)
{
    int flags = fcntl(ns->listen_fd, F_GETFL, 0);
    fcntl(ns->listen_fd, F_SETFL, flags | O_NONBLOCK);
    int peer = accept(ns->listen_fd, NULL, NULL);
    if (peer == -1) {
        perror("accept");
    } else {
        printf("new connection.\n");
        struct net_session *session = session_new(peer, SESSION_SERVER,
                ns->rsize, ns->wsize, ns->proc);
        if (session != NULL) {
            list_add_tail(&(session->list), &(ns->sessions)); /* 新的会话 */
            session->state = SESSION_INIT;
            thread_pool_add_work(net->pool, _users_work, session);
        } else {
            close(peer);
        }
    }
    fcntl(ns->listen_fd, F_SETFL, flags);
}

/* 处理监听套接字的可读，即接收新连接并创建新会话 */
static inline void _do_server_rw(fd_set *rset, fd_set *wset,
        struct list_head *s, struct net_manager* net)
{
    struct list_head *pos;
    list_for_each(pos, s)
    {
        struct net_server *ns = list_entry(pos, struct net_server, list);
        if (0 != lock_server(ns))
            continue;

        if (FD_ISSET(ns->listen_fd, rset))
            _do_server_accept(ns, net);

        /* 内部可能有链接断开，但此时服务器已经被上锁了 */
        _do_session_rw(rset, wset, &(ns->sessions), net);

        unlock_server(ns);
    }
}

static void read_write_fds(fd_set *rset, fd_set *wset,
        struct net_manager *manager)
{
    /* 不会改变服务端链表，使用读锁 */
    pthread_rwlock_rdlock(&(manager->serverslock));
    _do_server_rw(rset, wset, &(manager->servers), manager);
    pthread_rwlock_unlock(&(manager->serverslock));

    /* 连接断开时可能改变客户端链表，使用写锁 */
    pthread_rwlock_wrlock(&(manager->clientslock));
    _do_session_rw(rset, wset, &(manager->clients), manager);
    pthread_rwlock_unlock(&(manager->clientslock));
}

/* 处理所有网络连接中的可读的数据
 * 1、创建读的fd_set 和写的fd_set, 并将manager中的套接字设置到这两个集合中
 * 2、select
 * 3、检查返回的fd_set然后进行读、写
 */
static void _do_read_write(struct net_manager *manager)
{
    fd_set rfds, wfds;
    int max_fd = -1;

    set_rw_fds(&rfds, &wfds, &max_fd, manager);

    if (max_fd == -1) {
        usleep(1000 * 20);
        return;
    }

    int ret;

    // TODO 这个值影响新的会话被加入select的实时性, 需要一个合理的值
    struct timeval timeout = { 0, 200 * 1000 };

    /* 此处设置超时是因为，外部可能创建了新的连接需要select, 而不能一直阻塞在这里 */
    ret = select(max_fd + 1, &rfds, &wfds, NULL, &timeout);

    if (ret > 0) {
        read_write_fds(&rfds, &wfds, manager);
    } else if (ret < 0) {
        usleep(1000 * 20);
        perror("read select");
    } else {
        usleep(1000 * 20);
        // printf("no data for read within 100ms.\n");
    }

}

/*----------------- READ END ---------------------------*/
